/*
Tencent is pleased to support the open source community by making PhxQueue available.
Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
Licensed under the BSD 3-Clause License (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

<https://opensource.org/licenses/BSD-3-Clause>

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
*/



/* lock_client.cpp

 Generated by phxrpc_pb2client from lock.proto

*/

#include <cstdlib>
#include <iostream>
#include <memory>
#include <mutex>

#include "phxrpc/http.h"
#include "phxrpc/rpc.h"

#include "phxqueue/comm.h"

#include "lock_client_uthread.h"
#include "phxrpc_lock_stub.h"


using namespace std;


static phxrpc::ClientConfig global_lockclientuthread_config_;
static phxrpc::ClientMonitorPtr global_lockclientuthread_monitor_;


bool LockClientUThread::Init(const char *config_file) {
    return global_lockclientuthread_config_.Read(config_file);
}

const char *LockClientUThread::GetPackageName() {
    const char *ret = global_lockclientuthread_config_.GetPackageName();
    if (strlen(ret) == 0) {
        ret = "phxqueue_phxrpc.lock";
    }
    return ret;
}

LockClientUThread::LockClientUThread(phxrpc::UThreadEpollScheduler *uthread_scheduler) {
    uthread_scheduler_ = uthread_scheduler;
    static mutex monitor_mutex;
    if (!global_lockclientuthread_monitor_.get()) {
        monitor_mutex.lock();
        if (!global_lockclientuthread_monitor_.get()) {
            global_lockclientuthread_monitor_ = phxrpc::MonitorFactory::GetFactory()
                ->CreateClientMonitor(GetPackageName());
        }
        global_lockclientuthread_config_.SetClientMonitor(global_lockclientuthread_monitor_);
        monitor_mutex.unlock();
    }
}

LockClientUThread::~LockClientUThread() {}

int LockClientUThread::PHXEcho(const google::protobuf::StringValue &req,
                               google::protobuf::StringValue *resp) {
    const phxrpc::Endpoint_t *ep{global_lockclientuthread_config_.GetRandom()};

    if (uthread_scheduler_ != nullptr && ep != nullptr) {
        phxrpc::UThreadTcpStream socket;
        bool open_ret = phxrpc::PhxrpcTcpUtils::Open(uthread_scheduler_, &socket, ep->ip, ep->port,
                global_lockclientuthread_config_.GetConnectTimeoutMS(),
                *(global_lockclientuthread_monitor_.get()));
        if (open_ret) {
            socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());

            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
            return stub.PHXEcho(req, resp);
        }
    }

    return -1;
}

int LockClientUThread::PHXBatchEcho(const google::protobuf::StringValue &req,
                                    google::protobuf::StringValue *resp) {
    int ret = -1;
    size_t echo_server_count = 2;
    uthread_begin;
    for (size_t i = 0; i < echo_server_count; i++) {
        uthread_t [=, &uthread_s, &ret](void *) {
            const phxrpc::Endpoint_t *ep = global_lockclientuthread_config_.GetByIndex(i);
            if (ep != nullptr) {
                phxrpc::UThreadTcpStream socket;
                if (phxrpc::PhxrpcTcpUtils::Open(&uthread_s, &socket, ep->ip, ep->port,
                        global_lockclientuthread_config_.GetConnectTimeoutMS(),
                        *(global_lockclientuthread_monitor_.get()))) {
                    socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());
                    phxrpc::HttpMessageHandlerFactory http_msg_factory;
                    LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
                    int this_ret = stub.PHXEcho(req, resp);
                    if (this_ret == 0) {
                        ret = this_ret;
                        uthread_s.Close();
                    }
                }
            }
        };
    }
    uthread_end;
    return ret;
}

int LockClientUThread::GetString(const phxqueue::comm::proto::GetStringRequest &req,
                                 phxqueue::comm::proto::GetStringResponse *resp) {
    const phxrpc::Endpoint_t *ep{global_lockclientuthread_config_.GetRandom()};

    if (uthread_scheduler_ != nullptr && ep != nullptr) {
        phxrpc::UThreadTcpStream socket;
        bool open_ret = phxrpc::PhxrpcTcpUtils::Open(uthread_scheduler_, &socket, ep->ip, ep->port,
                    global_lockclientuthread_config_.GetConnectTimeoutMS(),
                    *(global_lockclientuthread_monitor_.get()));
        if (open_ret) {
            socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());

            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
            return stub.GetString(req, resp);
        }
    }

    return -1;
}

int LockClientUThread::SetString(const phxqueue::comm::proto::SetStringRequest &req,
                                 phxqueue::comm::proto::SetStringResponse *resp) {
    const phxrpc::Endpoint_t *ep{global_lockclientuthread_config_.GetRandom()};

    if (uthread_scheduler_ != nullptr && ep != nullptr) {
        phxrpc::UThreadTcpStream socket;
        bool open_ret = phxrpc::PhxrpcTcpUtils::Open(uthread_scheduler_, &socket, ep->ip, ep->port,
                    global_lockclientuthread_config_.GetConnectTimeoutMS(),
                    *(global_lockclientuthread_monitor_.get()));
        if (open_ret) {
            socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());

            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
            return stub.SetString(req, resp);
        }
    }

    return -1;
}

int LockClientUThread::DeleteString(const phxqueue::comm::proto::DeleteStringRequest &req,
                                    phxqueue::comm::proto::DeleteStringResponse *resp) {
    const phxrpc::Endpoint_t *ep{global_lockclientuthread_config_.GetRandom()};

    if (uthread_scheduler_ != nullptr && ep != nullptr) {
        phxrpc::UThreadTcpStream socket;
        bool open_ret = phxrpc::PhxrpcTcpUtils::Open(uthread_scheduler_, &socket, ep->ip, ep->port,
                    global_lockclientuthread_config_.GetConnectTimeoutMS(),
                    *(global_lockclientuthread_monitor_.get()));
        if (open_ret) {
            socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());

            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
            return stub.DeleteString(req, resp);
        }
    }

    return -1;
}

int LockClientUThread::GetLockInfo(const phxqueue::comm::proto::GetLockInfoRequest &req,
                                   phxqueue::comm::proto::GetLockInfoResponse *resp) {
    const phxrpc::Endpoint_t *ep{global_lockclientuthread_config_.GetRandom()};

    if (uthread_scheduler_ != nullptr && ep != nullptr) {
        phxrpc::UThreadTcpStream socket;
        bool open_ret = phxrpc::PhxrpcTcpUtils::Open(uthread_scheduler_, &socket, ep->ip, ep->port,
                    global_lockclientuthread_config_.GetConnectTimeoutMS(),
                    *(global_lockclientuthread_monitor_.get()));
        if (open_ret) {
            socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());

            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
            return stub.GetLockInfo(req, resp);
        }
    }

    return -1;
}

int LockClientUThread::AcquireLock(const phxqueue::comm::proto::AcquireLockRequest &req,
                                   phxqueue::comm::proto::AcquireLockResponse *resp) {
    const phxrpc::Endpoint_t *ep{global_lockclientuthread_config_.GetRandom()};

    if (uthread_scheduler_ != nullptr && ep != nullptr) {
        phxrpc::UThreadTcpStream socket;
        bool open_ret = phxrpc::PhxrpcTcpUtils::Open(uthread_scheduler_, &socket, ep->ip, ep->port,
                    global_lockclientuthread_config_.GetConnectTimeoutMS(),
                    *(global_lockclientuthread_monitor_.get()));
        if (open_ret) {
            socket.SetTimeout(global_lockclientuthread_config_.GetSocketTimeoutMS());

            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            LockStub stub(socket, *(global_lockclientuthread_monitor_.get()), http_msg_factory);
            return stub.AcquireLock(req, resp);
        }
    }

    return -1;
}

